package com.example.java_chatroom.listener;

import com.example.java_chatroom.Constants.CacheConstants;
import com.example.java_chatroom.Constants.Constants;
import com.example.java_chatroom.Constants.RabbitMqConstants;
import com.example.java_chatroom.model.*;
import com.example.java_chatroom.component.OnlineUserManager;
import com.example.java_chatroom.redis.service.RedisService;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Component
public class MessageListener {
    private static final Logger log = LoggerFactory.getLogger(MessageListener.class);
    @Autowired
    private ObjectMapper objectMapper;
    @Autowired
    private MessageSessionMapper messageSessionMapper;
    @Autowired
    private OnlineUserManager onlineUserManager;
    @Autowired
    private MessageMapper messageMapper;
    @Autowired
    private RedisService redisService;
    //持久化用的线程池
    private static ExecutorService saveExecutorService = Executors.newFixedThreadPool(2);
    //转发用的线程池
    private static ExecutorService transferExecutorService = Executors.newFixedThreadPool(5);

    //缓存消息
    private static ConcurrentLinkedQueue<com.example.java_chatroom.model.Message> messageQueue = new ConcurrentLinkedQueue<>();

    @RabbitListener(queues = RabbitMqConstants.WORK_QUEUE_NAME)
    public void queueListener(Message message, Channel channel) throws IOException {
        //确认机制
        log.error(Long.toString(message.getMessageProperties().getDeliveryTag()));
        // 处理消息
        String msg = new String(message.getBody(), "UTF-8");
        System.out.println("Received message: " + msg);

        // 获取消息内容 (消息体)
        MessageRequest req = objectMapper.readValue(msg, MessageRequest.class);
        //在这里再处理消息转发和持久化
        if (req.getType().equals("message")) {
            // 验证消息的类型，如果是message就转发
            transferExecutorService.submit(() -> {
                try {
                    transferMessage(req.getUser(), req);
                    // 手动确认消息
                    long deliveryTag = message.getMessageProperties().getDeliveryTag();
                    channel.basicAck(deliveryTag, false); // 确认消息
                } catch (IOException e) {
                    //如果不是就抛出异常
                    // 处理失败，否定确认消息并重新入队
                    long deliveryTag = message.getMessageProperties().getDeliveryTag();
                    try {
                        channel.basicNack(deliveryTag, false, true); // 将消息重新放回队列
                    } catch (IOException ex) {
                        throw new RuntimeException(ex);
                    }
                    System.out.println("Message processing failed: " + e.getMessage());
                    throw new RuntimeException(e);
                }

            });
        } else {
            throw new RuntimeException();
        }


    }

    // 通过这个方法来完成消息实际的转发工作.
    // 第一个参数就表示这个要转发的消息, 是从谁那来的.
    private void transferMessage(User fromUser, MessageRequest req) throws IOException {
        // 1. 先构造一个待转发的响应对象. MessageResponse
        MessageResponse resp = new MessageResponse();
        resp.setType("message"); // 这里不设置也行, 默认也就是 message
        resp.setFromId(fromUser.getUserId());
        resp.setFromName(fromUser.getUsername());
        resp.setSessionId(req.getSessionId());
        resp.setContent(req.getContent());
        // 把这个 java 对象转成 json 格式字符串
        String respJson = objectMapper.writeValueAsString(resp);
        System.out.println("[transferMessage] respJson: " + respJson);

        // 2. 根据请求中的 sessionId, 获取到这个 MessageSession 里都有哪些用户. 通过查询数据库就能够知道了.
        //直接从redis中取
        if (redisService.getListSize(getSessionKey(req.getSessionId())) == 0) {
            refreshRedis();
        }
        List<Integer> friendIdList = redisService.getCacheListByRange(getSessionKey(req.getSessionId()), 0, -1, Integer.class);


        // 3. 循环遍历上述的这个列表, 给列表中的每个用户都发一份响应消息
        //    注意: 这里除了给查询到的好友们发, 也要给自己也发一个. 方便实现在自己的客户端上显示自己发送的消息.
        //    注意: 一个会话中, 可能有多个用户(群聊). 虽然客户端是没有支持群聊的(前端写起来相对麻烦), 后端无论是 API 还是 数据库
        //          都是支持群聊的. 此处的转发逻辑也一样让它支持群聊.
        for (Integer friendId : friendIdList) {
            // 知道了每个用户的 userId, 进一步的查询刚才准备好的 OnlineUserManager, 就知道了对应的 WebSocketSession
            // 从而进行发送消息
            WebSocketSession webSocketSession = onlineUserManager.getSession(friendId);
            if (webSocketSession == null) {
                // 如果该用户未在线, 则不发送.
                continue;
            }
            webSocketSession.sendMessage(new TextMessage(respJson));
        }

        // 4. 消息持久化
        com.example.java_chatroom.model.Message message = new com.example.java_chatroom.model.Message();
        message.setFromId(fromUser.getUserId());
        message.setSessionId(req.getSessionId());
        message.setContent(req.getContent());
        //先保存到内存里，到达一定数据量再一块存到数据库
        messageQueue.add(message);
    }

    //定时持久化消息
    // 定时任务：每隔1秒扫描一次消息列表
    // 定时持久化消息，批量保存到数据库
    @Scheduled(fixedRate = 1000) // 每1秒执行一次
    private void saveMessage() throws IOException {
        saveExecutorService.submit(() -> {
            List<com.example.java_chatroom.model.Message> batch = new ArrayList<>();
            // 从队列中取出一定数量的消息进行持久化
            for (int i = 0; i < Constants.SAVE_COUNT; i++) {
                com.example.java_chatroom.model.Message message = messageQueue.poll();
                if (message == null) {
                    break;  // 如果队列为空，则跳出
                }
                batch.add(message);
            }

            if (!batch.isEmpty()) {
                messageMapper.addAll(batch);
            }
        });
    }

    private String getSessionKey(int sessionId) {
        return CacheConstants.CACHE_SESSION + sessionId;
    }

    private void refreshRedis() {
        //先从message_session_user表中查询所有记录
            List<MessageSessionUserItem> messageSessionUserItemList = messageSessionMapper.getAllMessageSessionUsers();
        //遍历messageSessionUserItemList，封装出一个哈希表
        ConcurrentHashMap<Integer, List<Integer>> concurrentHashMap = new ConcurrentHashMap<>();
        for (MessageSessionUserItem messageSessionUserItem : messageSessionUserItemList) {
            if (concurrentHashMap.containsKey(messageSessionUserItem.getSessionId())) {
                concurrentHashMap.get(messageSessionUserItem.getSessionId()).add(messageSessionUserItem.getUserId());
            } else {
                List<Integer> friendIdList = new ArrayList<>();
                friendIdList.add(messageSessionUserItem.getUserId());
                concurrentHashMap.put(messageSessionUserItem.getSessionId(),friendIdList );
            }
        }
        //遍历哈希表
        // 使用 entrySet() 遍历哈希表
        for (Map.Entry<Integer, List<Integer>> entry : concurrentHashMap.entrySet()) {
            Integer sessionId = entry.getKey();
            List<Integer> userIds = entry.getValue();
            // 处理每个 sessionId 和其对应的 userId 列表
            redisService.rightPushAll(getSessionKey(sessionId), userIds);
        }
    }
}